package join;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ClickJoinUserSimple extends Configured implements Tool {

	private static final String userSeparator = "\u0001";
	private static final String clickSeparator = "\t";
	private static HashMap<String, String> clickSortMap = null;

	public static boolean isValid(String src) {
		return (src != null && !src.equals("") && !src.equals("-"));
	}

	public static void sortFilterSetUp() {
		clickSortMap = new HashMap<String, String>();
		// bid,reserve_price,price_asc ps=reserve_price&ss=ends price_asc
		// 按照价格正序
		clickSortMap.put("bid", "price_asc");
		clickSortMap.put("reserve_price", "price_asc");
		clickSortMap.put("price_asc", "price_asc");
		clickSortMap.put("ps=reserve_price&ss=ends", "price_asc");
		// _bid,_reserve_price,price_desc _ps=reserve_price&ss=ends
		// price_desc 按照价格逆序
		clickSortMap.put("_bid", "price_desc");
		clickSortMap.put("_reserve_price", "price_desc");
		clickSortMap.put("price_desc", "price_desc");
		clickSortMap.put("_ps=reserve_price&ss=ends", "price_desc");
		// coefp ps=coefp&ss=ends coefp_desc 人气排序正序
		clickSortMap.put("coefp", "coefp_desc");
		clickSortMap.put("ps=coefp&ss=ends", "coefp_desc");
		// _coefp _ps=coefp&ss=ends coefp_asc 人气排序逆序
		clickSortMap.put("_coefp", "coefp_asc");
		clickSortMap.put("_ps=coefp&ss=ends", "coefp_asc");
		// _isnewest, isnewest _ps=isnewest&ss=ends, ps=isnewest&ss=ends
		// isnewest 最新上架产品优先
		clickSortMap.put("_isnewest", "isnewest");
		clickSortMap.put("isnewest", "isnewest");
		clickSortMap.put("_ps=isnewest&ss=ends", "isnewest");
		clickSortMap.put("ps=isnewest&ss=ends", "isnewest");
		// ratesum, credit_asc ps=ratesum&ss=ends ratesum_asc
		clickSortMap.put("ratesum", "ratesum_asc");
		clickSortMap.put("credit_asc", "ratesum_asc");
		clickSortMap.put("ps=ratesum&ss=ends", "ratesum_asc");
		// _ratesum，credit_desc _ps=ratesum&ss=ends ratesum_desc
		clickSortMap.put("_ratesum", "ratesum_desc");
		clickSortMap.put("credit_desc", "ratesum_desc");
		clickSortMap.put("_ps=ratesum&ss=ends", "ratesum_desc");
		// sale_asc ps=volume30day&ss=ends sale_asc
		clickSortMap.put("sale_asc", "sale_asc");
		clickSortMap.put("ps=volume30day&ss=ends", "sale_asc");
		// sale_desc _ps=volume30day&ss=ends sale_desc
		clickSortMap.put("sale_desc", "sale_desc");
		clickSortMap.put("_ps=volume30day&ss=ends", "sale_desc");
		// _ends, _all, _commend _ps=commend&ss=ends,
		// _ps=ends&ss=reserve_price, _ps=assure&_ss=commend all
		clickSortMap.put("_ends", "all");
		clickSortMap.put("_all", "all");
		clickSortMap.put("_commend", "all");
		clickSortMap.put("_ps=commend&ss=ends", "all");
		clickSortMap.put("_ps=ends&ss=reserve_price", "all");
		clickSortMap.put("_ps=assure&_ss=commend", "all");
		// end, all,commend ps=ends&ss=reserve_price,ps=commend&ss=ends,
		// ps=assure&_ss=commend all
		clickSortMap.put("end", "all");
		clickSortMap.put("all", "all");
		clickSortMap.put("commend", "all");
		clickSortMap.put("ps=ends&ss=reserve_price", "all");
		clickSortMap.put("ps=commend&ss=ends", "all");
		clickSortMap.put("ps=assure&_ss=commend", "all");
	}

	private static String sortFilt(String src) {
		return clickSortMap.containsKey(src) ? clickSortMap.get(src) : "other";
	}

	public static class UserMap extends MapReduceBase implements
			Mapper<BytesWritable, Text, Text, Text> {
		Text keyPair = new Text();
		String line = new String("");
		Text value = new Text();

		public void map(BytesWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			line = value.toString();
			String[] user = line.split(userSeparator, 18);
			if (user.length == 18 && isValid(user[17])) {
				keyPair.set(user[17]);
				value.set(user[0]);
				// System.out.println("UserMap: " + keyPair.toString());
				output.collect(keyPair, value);
			}
		}

	}

	public static class ClickMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		Text keyPair = new Text();
		String line = new String("");
		Text value = new Text();

		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			sortFilterSetUp();
			line = value.toString();
			String[] click = line.split(clickSeparator, 27);
			if (click.length == 27 && click[0].equals("c1")) {
				if (isValid(click[3])) {
					keyPair.set(click[3]);
					value.set(line + clickSeparator + sortFilt(click[11]));
					// System.out.println("ClickMap: " + keyPair.toString());
					output.collect(keyPair, value);
				}
			}
		}
	}

	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			ArrayList<String> container = new ArrayList<String>();
			String userId = "";
			while (values.hasNext()) {
				String v = values.next().toString();
				if (v.contains("\t"))
					container.add(v);
				else
					userId = v;
			}
			if (isValid(userId)) {
				System.out.println("userID: " + userId);
				for (String s : container)
					output.collect(new Text(userId), new Text(s));
			}
		}
	}

	public int run(String[] args) throws Exception {
		String date = args[0];
		// Click
		String clickPathStr = "/group/taobao/taobao/dw/fact/" + date
				+ "/NewSearchLog/";
		Path clickPath = new Path(clickPathStr);
		// User
		String userPathStr = "/group/taobao/taobao/hive/s_bmw_users_extra/pt=20100318000000/";
		Path userPath = new Path(userPathStr);
		// Output
		String clickUserPathStr = "/group/tbsc-dev/xzwang/cyliu/join/intermediate/"
				+ date + "/clickUser";
		Path clickUserPath = new Path(clickUserPathStr);
		JobConf conf = new JobConf(getConf(), ClickJoinUserSimple.class);
		conf.setJobName("ClickJoinUser");
		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(Text.class);

		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, userPath,
				SequenceFileInputFormat.class, UserMap.class);
		MultipleInputs.addInputPath(conf, clickPath, TextInputFormat.class,
				ClickMap.class);
		FileOutputFormat.setOutputPath(conf, clickUserPath);
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(),
				new ClickJoinUserSimple(), args);
		System.exit(res);
	}

}
